package cn.seecoder.ai.service

import cn.seecoder.ai.dao.FileInfoRepository
import cn.seecoder.ai.utils.{DataAnalysisUtil, HdfsHelper}
import org.apache.spark.sql.SparkSession
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service

/**
 * @author DingXiaoyu
 * @date 2023/4/19 20:37
 */
@Service
class FileAnalysisService {

  @Autowired
  var sparkSession: SparkSession = _

  @Autowired
  var hdfsHelper: HdfsHelper = _

  @Autowired
  var fileInfoRepository: FileInfoRepository = _

  @Autowired
  var dataAnalysisUtil: DataAnalysisUtil = _


  def dataDistribution(fileInfoId:Integer,chosenColumn:String,dataType:String):java.util.Map[String,Integer] = {
    val uri=fileInfoRepository.findById(fileInfoId).get().getOriginalFileUri
    val df = sparkSession.read
      .option("header", "true")
      .option("inferSchema", "true")
      .csv(uri)
      .persist()
    if (dataType.equals("integer")){
      dataAnalysisUtil.intDataDistribution(df.select(chosenColumn).collectAsList())
    }else if (dataType.equals("double")){
      dataAnalysisUtil.doubleDataDistribution(df.select(chosenColumn).collectAsList())
    }else if (dataType.equals("string")){
      dataAnalysisUtil.stringDataDistribution(df.select(chosenColumn).collectAsList())
    }else{
      null
    }
  }

  def scatterPlot(fileInfoId:Integer,firstChosenColumn:String,secondChosenColumn:String):java.util.List[java.util.Map[String,java.lang.Double]]={
    val uri = fileInfoRepository.findById(fileInfoId).get().getOriginalFileUri
    val df = sparkSession.read
      .option("header", "true")
      .option("inferSchema", "true")
      .csv(uri)
      .persist()
    dataAnalysisUtil.scatterPlot(df.select(firstChosenColumn,secondChosenColumn).collectAsList(),firstChosenColumn,secondChosenColumn)
  }

  def statistics(fileInfoId:Integer,chosenColumn:String):java.util.Map[String,java.lang.Double]={
    val uri = fileInfoRepository.findById(fileInfoId).get().getOriginalFileUri
    val df = sparkSession.read
      .option("header", "true")
      .option("inferSchema", "true")
      .csv(uri)
      .persist()
    dataAnalysisUtil.statistics(df.select(chosenColumn).collectAsList())
  }

  def skewnessKurtosis(fileInfoId:Integer,chosenColumn:String):java.util.Map[String,java.lang.Double]={
    val uri = fileInfoRepository.findById(fileInfoId).get().getOriginalFileUri
    val df = sparkSession.read
      .option("header", "true")
      .option("inferSchema", "true")
      .csv(uri)
      .persist()
    dataAnalysisUtil.skewnessKurtosis(df.select(chosenColumn).collectAsList())
  }

  def pearson(fileInfoId: Integer, firstChosenColumn: String, secondChosenColumn: String): java.util.Map[String, java.lang.Double] = {
    val uri = fileInfoRepository.findById(fileInfoId).get().getOriginalFileUri
    val df = sparkSession.read
      .option("header", "true")
      .option("inferSchema", "true")
      .csv(uri)
      .persist()
    dataAnalysisUtil.pearson(df.select(firstChosenColumn,secondChosenColumn).collectAsList())
  }

  def getRows(path: String, label:String): java.util.List[org.apache.spark.sql.Row]={
    val df = sparkSession.read
      .option("header", "true")
      .option("inferSchema", "true")
      .csv(path)
      .persist()
    df.select(label).collectAsList()
  }

  def test(): Unit = {
    val df = sparkSession.read
      .option("header", "true")
      .option("inferSchema", "true")
      .csv("C:\\Users\\Lenovo\\Desktop\\a.csv")
      .persist()

    println(dataAnalysisUtil.pearson(df.select("年龄","数字").collectAsList()))
  }
}
